package org.jef.socket.websocket;

import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.jef.security.service.PigUser;
import org.jef.socket.event.EventType;
import org.jef.socket.msg.IMsg;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @version 1.0
 * @description: 抽象websocket,核心校验功能
 * @author: ChengZhi 0336
 * @create: 2024-08-23 10:27
 **/
@Slf4j
@ServerEndpoint("/globalWs/{token}/{roomId}")
public abstract class AbstractWebSocket implements IWebSocket{

    /**
     * 配置文件中配置幂等过期时间
     */
    @Value("${jef.websocket.model}")
    private String model;

    private final static String DEFAULT_MODLE = "single";

    private static final Map<String, Map<Long, AbstractWebSocket>> CLIENTS = new ConcurrentHashMap<String, Map<Long, AbstractWebSocket>>();


    protected Session session;

    protected String token;

    private Long userId;

    private String roomId;

    protected PigUser pigUser;

    /**
     * 验证token是否有效
     *
     * @param token  token
     * @param isInit 是否对token和userInfo进行初始化赋值
     * @return boolean
     */
    protected abstract boolean isTokenValid(String token, boolean isInit);

    /**
     * 认证失败处理
     *
     * @param session session
     */
    protected abstract void sendAuthFailed(Session session);

    /**
     * 当接收到客户端消息后是否自动回复
     * @return
     */
    protected abstract boolean isAutoReply();

    protected abstract PigUser getPigUser();

    @OnOpen
    public void onOpen(Session session, @PathParam("token") String token, @PathParam("roomId") String roomId) {
        if (!isTokenValid(token, true)) {
            sendAuthFailed(session);
            return;
        }
        pigUser = getPigUser();
        this.session = session;
        this.userId = pigUser.getId();

        Map<Long, AbstractWebSocket> room = CLIENTS.get(roomId);
        if (room == null) {
            room = new ConcurrentHashMap<Long, AbstractWebSocket>();
            room.put(userId, this);
        } else {
            room.put(userId, this);
        }
        CLIENTS.put(roomId, room);
        log.info("当前房间{}用户:{}",roomId, room.keySet());

        try {
            session.getBasicRemote().sendText(userId + "连接成功！");
        } catch (IOException e) {
            log.error("cokcet连接失败", e);
        }
    }

    /**
     * 收到客户端消息后调用的方法
     * @param session
     * @param message
     * @return
     */
    @OnMessage
    public String onMessage(Session session, String message) {
        try {
            if (isAutoReply()) {
                session.getBasicRemote().sendText("received");
            }
        } catch (IOException e) {
            log.error("自动回复已收到消息异常", e);
            e.printStackTrace();
        }
        return null;
    }


    @OnError
    public void onError(Session session, Throwable throwable) {
        log.error("ws session 发生错误:{}", throwable.getMessage());
    }

    /**
     * 客户端关闭触发
     * @param session
     */
    @OnClose
    public void onClose(Session session) {
        CLIENTS.get(roomId).remove(userId);
        log.info("ws 用户 userKey {} 已下线,当前在线用户:{}", userId, CLIENTS.get(roomId).keySet());
    }

    /**
     * 发送消息
     *
     * @param msg
     */
    public void sendMessage(IMsg msg) {
        try {
            this.session.getBasicRemote().sendText(msg.getMsg());
        } catch (IOException e) {
            log.error("发送消息异常", e);
        }
    }

    /**
     * 发送消息
     * @param event
     * @param msg
     */
    public void send(EventType event, IMsg msg) {
        // 根据配置处理消息
        if (model != null && DEFAULT_MODLE.equals(model)) {
            Long userId = msg.getUserId();
            String roomId = msg.getRoomId();
            if (userId != null && userId != 0 && roomId != null && !"".equals(roomId)) {
                // userId不为0说明是点对点消息
                this.sendToUser(roomId, userId, msg);
            } else if (roomId != null && !"".equals(roomId) && (userId == null || userId == 0)) {
                // 房间内广播
                this.roomBroadcast(msg, roomId);
            } else {
                // 广播消息
                this.broadcast(msg);
            }
        } else {
            RedisTemplate template = SpringUtil.getBean("jefRedisTemplate");
            template.convertAndSend(event.getValue(), msg);
        }
    }

    /**
     * 批量发送消息
     * @param event
     * @param msgs
     */
    public void sendBatch(EventType event, List<IMsg> msgs) {
        if (model != null && DEFAULT_MODLE.equals(model)) {
            throw new RuntimeException("暂时不支持批量发送！");
        } else {
            RedisTemplate template = SpringUtil.getBean("jefRedisTemplate");
            template.executePipelined((RedisCallback<Object>) connection -> {
                for (IMsg message : msgs) {
                    template.convertAndSend(event.getValue(), message);
                }
                return null;
            });
        }
    }

    /**
     * 向user精确用户发送消息
     *
     * @param roomId        房间编号
     * @param userId        用户编号
     * @param msg           消息
     */
    public void sendToUser(String roomId, Long userId, IMsg msg) {
        printClientsInfo();
        Map<Long, AbstractWebSocket> room = CLIENTS.get(roomId);
        if (room != null) {
            AbstractWebSocket webSocket = room.get(userId);
            if (null != webSocket) {
                webSocket.sendMessage(msg);
                return;
            }
        }
        log.error("发送消息到指定用户,但是用户不存在,userId is {},message is {}", userId, msg.getMsg());
    }

    /**
     * 全体广播
     * @param
     */
    public void broadcast(IMsg msg) {

        printClientsInfo();
        for (Map<Long, AbstractWebSocket> room : CLIENTS.values()) {

            if (room != null) {
                for (Map.Entry entry : room.entrySet()) {
                    AbstractWebSocket abstractWebSocket = (AbstractWebSocket) entry.getValue();
                    Session curSession = abstractWebSocket.session;
                    try {
                        curSession.getBasicRemote().sendText(msg.getMsg());
                    } catch (IOException e) {
                        log.error("发送ws数据错误:{}", e.getMessage());
                    }
                }
            }
        }
    }

    /**
     * 房间内广播
     * @param msg
     */
    public void roomBroadcast(IMsg msg, String roomId) {
        printClientsInfo();
        Map<Long, AbstractWebSocket> room = CLIENTS.get(roomId);
        if (room != null) {
            for (AbstractWebSocket abstractWebSocket : room.values()) {
                Session curSession = abstractWebSocket.session;
                if (curSession.isOpen()) {
                    try {
                        curSession.getBasicRemote().sendText(msg.getMsg());
                    } catch (IOException e) {
                        log.error("发送ws数据错误:{}", e.getMessage());
                    }
                }
            }
        }
    }

    private void printClientsInfo() {
        if (log.isDebugEnabled()) {
            log.debug("print global info start:" );
            for (String roomId : CLIENTS.keySet()) {
                for (Long userId : CLIENTS.get(roomId).keySet()) {
                    log.debug("roomId:{}   userId:{}",roomId, userId);
                }
            }
            log.debug("print global info end!" );
        }
    }
}
